package baiduschPro;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import org.apache.nifi.annotation.behavior.ReadsAttribute;
import org.apache.nifi.annotation.behavior.ReadsAttributes;
import org.apache.nifi.annotation.behavior.WritesAttribute;
import org.apache.nifi.annotation.behavior.WritesAttributes;
import org.apache.nifi.annotation.documentation.CapabilityDescription;
import org.apache.nifi.annotation.documentation.SeeAlso;
import org.apache.nifi.annotation.documentation.Tags;
import org.apache.nifi.annotation.lifecycle.OnScheduled;
import org.apache.nifi.components.PropertyDescriptor;
import org.apache.nifi.components.PropertyValue;
import org.apache.nifi.flowfile.FlowFile;
import org.apache.nifi.processor.AbstractProcessor;
import org.apache.nifi.processor.ProcessContext;
import org.apache.nifi.processor.ProcessSession;
import org.apache.nifi.processor.ProcessorInitializationContext;
import org.apache.nifi.processor.Relationship;
import org.apache.nifi.processor.exception.ProcessException;
import org.apache.nifi.processor.io.OutputStreamCallback;
import org.apache.nifi.processor.util.StandardValidators;

import hylanda.common.save2MysqlBase;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;

@Tags({"get data from mysql "})
@CapabilityDescription("get data from mysql,input")
@SeeAlso({})
@ReadsAttributes({@ReadsAttribute(attribute="", description="")})
@WritesAttributes({@WritesAttribute(attribute="", description="")})

public class GetDataFromMysql extends AbstractProcessor {

	
	 public static final PropertyDescriptor MYSQLURL = new PropertyDescriptor
	            .Builder().name("MYSQLURL")
	            .displayName("mysqlurl")
	            .description("mysql url")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSQLUSER = new PropertyDescriptor
	            .Builder().name("MYSQLUSER")
	            .displayName("mysqluser")
	            .description("mysql user")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSQLPWD = new PropertyDescriptor
	            .Builder().name("MYSQLPWD")
	            .displayName("mysqlpwd")
	            .description("mysql password")
	            .required(false)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final PropertyDescriptor MYSQLSQL = new PropertyDescriptor
	            .Builder().name("MYSQLSQL")
	            .displayName("mysqlsql")
	            .description("mysql sql")
	            .required(true)
	            .addValidator(StandardValidators.NON_EMPTY_VALIDATOR)
	            .build();
	 
	 public static final Relationship RELATIONSHIP_OK = new Relationship.Builder()
	            .name("gen success")
	            .description("success relationship")
	            .build();
	    
	 public static final Relationship RELATIONSHIP_NO = new Relationship.Builder()
	            .name("gen failure")
	            .description("failure relationship")
	            .build();
	
	    
	 private List<PropertyDescriptor> descriptors;
	 private Set<Relationship> relationships;
	    
	 private save2MysqlBase mysqlselect = new save2MysqlBase();
	    
	 @Override
	 protected void init(final ProcessorInitializationContext context) {
	        final List<PropertyDescriptor> descriptors = new ArrayList<PropertyDescriptor>();
	        descriptors.add(MYSQLURL);
	        descriptors.add(MYSQLUSER);
	        descriptors.add(MYSQLPWD);
	        descriptors.add(MYSQLSQL);
	        this.descriptors = Collections.unmodifiableList(descriptors);

	        final Set<Relationship> relationships = new HashSet<Relationship>();
	        relationships.add(RELATIONSHIP_OK);
	        relationships.add(RELATIONSHIP_NO);
	        this.relationships = Collections.unmodifiableSet(relationships);

	    }
	 
	public GetDataFromMysql() {
		// TODO Auto-generated constructor stub
	}

	
	@Override
    public Set<Relationship> getRelationships() {
        return this.relationships;
    }

    @Override
    public final List<PropertyDescriptor> getSupportedPropertyDescriptors() {
        return descriptors;
    }

    @OnScheduled
    public void onScheduled(final ProcessContext context) {

  //  	System.out.println("onScheduled is clled");
    }
	    
	@Override
	public void onTrigger(ProcessContext context, ProcessSession session) throws ProcessException {
		// TODO Auto-generated method stub
		
		PropertyValue mysqlurlp = context.getProperty("MYSQLURL");
    	PropertyValue mysqluserp= context.getProperty("MYSQLUSER");
    	PropertyValue mysqlpwdp = context.getProperty("MYSQLPWD");
    	PropertyValue mysqlsqlp = context.getProperty("MYSQLSQL");
       
        
       String mysqlurl =  mysqlurlp.getValue();
       String mysqluser = mysqluserp.getValue();
       String mysqlpwd = mysqlpwdp.getValue();
       String mysqlsql = mysqlsqlp.getValue();
       mysqlselect.setParams(mysqlurl,mysqluser,mysqlpwd);
       mysqlselect.initmysql();
       
       List lisdatta = mysqlselect.selectData(mysqlsql,"");
       if(lisdatta != null)
       {
	       for(int i = 0; i < lisdatta.size() ; i++)
	       {
	    	   FlowFile flowFile = session.create(); 
	    	   String strjson = (String) lisdatta.get(i);
	    	   final String data = strjson;
	    	   
	    	   final long importStart = System.nanoTime();
               
               
	    	   flowFile = session.write(flowFile, new OutputStreamCallback() {
	               @Override
	               public void process(final OutputStream out) throws IOException {
	                   out.write(data.getBytes("UTF-8"));
	               }
	           }); 
	  //  	   final long importNanos = System.nanoTime() - importStart;
        //       final long importMillis = TimeUnit.MILLISECONDS.convert(importNanos, TimeUnit.NANOSECONDS); 
          //     String strMessage = String.format("get No.%d", i);
            //   session.getProvenanceReporter().receive(flowFile,strMessage , importMillis); 
	    	   session.transfer(flowFile, RELATIONSHIP_OK); 
	       }// for i   
       }
       else
       {
    	   FlowFile flowFile = session.create(); 
    	   String data = "no data find!";
    	   flowFile = session.write(flowFile, new OutputStreamCallback() {
               @Override
               public void process(final OutputStream out) throws IOException {
                   out.write(data.getBytes("UTF-8"));
               }
           }); 
    	   session.transfer(flowFile, RELATIONSHIP_NO); 
       }
       
       mysqlselect.deconnSQL();
	}
}
